import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

public class UVReducer extends Reducer<Text, NullWritable, Text, IntWritable> {
    private Map<String, Integer> count = new HashMap<String, Integer>();

    @Override
    protected void reduce(Text key, Iterable<NullWritable> values, Context context)
            throws IOException, InterruptedException {
        String data = key.toString();
        String date = data.split("\t")[0];
        String ip = data.split("\t")[1];

        if (count.containsKey(date)) count.put(date, count.get(date) + 1);
        else count.put(date, 1);
    }

    @Override
    protected void cleanup(Reducer<Text, NullWritable, Text, IntWritable>.Context context)
            throws IOException, InterruptedException {
        Set<String> keyDate = count.keySet();
        for (String value : keyDate) {
            context.write(new Text(value), new IntWritable(count.get(value)));
        }
    }
}
